package com.yanggu.flink.datastream_api.window

import cn.hutool.core.date.DateUtil
import com.yanggu.flink.datastream_api.pojo.SensorReading
import com.yanggu.flink.datastream_api.source.MySensorSource
import org.apache.flink.configuration.{Configuration, RestOptions}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.util.Date

/**
 * 滑动窗口(Sliding Window): 滑动窗口是滚动窗口更广义的一种形式, 滑动窗口由固定的窗口长度和滑动步长组成
 * 一般而言滑动步长小于窗口大小
 * 特点: 时间长度对齐, 窗口长度固定, 可以有重叠
 * 滑动窗口分配器将元素分配到固定长度的窗口中, 与滚动窗口类似。窗口的大小由窗口大小参数来配置, 另一个窗口滑动参数控制滑动窗口开始的频率。因此， 滑动窗口如果滑动参数小于窗口大小的话，窗口是可以重叠的，在这种情况下元素 会被分配到多个窗口中
 * 适用场景: 统计过去5分钟内, 接口失败的次数, 每隔5秒钟输出一次, 适合高频输出的场景
 * SlidingEventTimeWindows.of(windowSize, slide, offset), 第一个参数是窗口大小, 第二个参数是滑动步长, 第三个参数是偏移量
 * 一个数据可能属于多个窗口, 窗口之间可能会有重叠
 */
object SlidingWindowDemo {

  def main(args: Array[String]): Unit = {
    //创建本地执行环境, 并且拥有WebUi和设置端口
    val config = new Configuration()
    config.setInteger(RestOptions.PORT.key(), 8080)
    val environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config)
    environment.setParallelism(1)

    environment
      .addSource(new MySensorSource)
      //设置事件时间升序的Watermark
      .assignAscendingTimestamps(_.timestamp)
      //根据id进行分区
      .keyBy(_.id)
      //创建了事件时间滑动窗口, 窗口大小为1minute, 滑动步长为5s
      .window(SlidingEventTimeWindows.of(Time.minutes(1L), Time.seconds(5L)))
      //执行全窗口函数, 将窗口内的数据集收集齐后执行process方法
      .process(new ProcessWindowFunction[SensorReading, String, String, TimeWindow] {
        override def process(key: String, context: Context, elements: Iterable[SensorReading], out: Collector[String]): Unit = {
          val result = s"当前时间: ${DateUtil.formatDateTime(new Date)}, " +
            s"滑动窗口开始时间: ${DateUtil.formatDateTime(new Date(context.window.getStart))}, 窗口结束时间: ${DateUtil.formatDateTime(new Date(context.window.getEnd))}, " +
            s"传感器id: $key" + s"过去1minute, 总共有${elements.size}条传感器数据"
          out.collect(result)
        }
      })
      .print("result ")

    environment.execute("TumblingWindowDemo Job")
  }

}
